package com.aotain.nyx.port;

import java.util.HashMap;

import org.apache.commons.codec.binary.Base64;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;

import com.aotain.apollo.ApolloConfig;
import com.aotain.apollo.PortInfo;
import com.aotain.nyx.flink.BasicTuple;

/**
 * 
        *@Title:检测端口异常MAP
        *@Description:
        *@Author:Turk
        *@Since:2017年6月30日
        *@Version:1.1.0
 */
public class PortMap implements MapFunction<Tuple2<String,String>,Tuple2<String,BasicTuple>>{
	
	/**
	  *
	  */
	    
	private static final long serialVersionUID = 6767296645263877879L;
	
	HashMap<Integer,PortInfo> ports = null;
	
	
	public PortMap(String dbJson) {
		//初始化
		ApolloConfig ap = new ApolloConfig(dbJson);
		
		//加载配置信息
		ports = ap.getPortThreaten();
		
		ports.put(22,new PortInfo(22, 80, "SSH远程协议访问",0));
		ports.put(21,new PortInfo(21, 80, "ftp访问",0));
		ports.put(23,new PortInfo(23, 80, "Telnet访问",0));
		ports.put(3389, new PortInfo(3389, 80, "Windows远程终端访问",0));
		//135、139、445、593、1025
		ports.put(135, new PortInfo(135, 80, "Windows DCOM服务访问",0));
		ports.put(139, new PortInfo(139, 80, "NetBIOS/SMB服务访问",0));
		ports.put(445, new PortInfo(445, 80, "公共Internet文件系统访问",0));
		ports.put(1433, new PortInfo(1433, 80, "SQL Server常用端口访问",0));
	}

	@Override
	public Tuple2<String, BasicTuple> map(Tuple2<String, String> value)
			throws Exception {
		
//		// [0-4]dip+ "|" +dport+ "|" +sip+ "|" +sport+ "|" +gisstrBase64+ "|" + 
		// [5-9]domainname+ "|" +url+ "|" +type+ "|" +delaytime+ "|" +idcno+ "|" + 
		// [10-14]upStreamOctets+ "|" +upStreamPacket+ "|" +status+ "|" +userAgent+ "|" +cookie+ "|" + 
		// [15-19]dnStreamOctets+ "|" +dnStreamPacket+ "|" + refer + "|" +post | OUT/IN
		
		Tuple2<String, BasicTuple> ret = new Tuple2<String, BasicTuple>();
		
		String logString = value.f1; //格式化后的原始数据
		String[] items = logString.split("\\|",-1);
		
		if(items.length < 12) {
			return ret;
		}
		
		String destip = items[0];
		int destport = Integer.parseInt(items[1]);
		String sourceip = items[2];
		byte[] gisbytes =  Base64.decodeBase64(items[4]);
		String gisstr = new String(gisbytes);
		String idc = items[9];
		String flowdirection = items[19];
		
		long upStreamOctets = 0; //UPSTREAMOCTETS
		long upStreamPacket = 0; //UPSTREAMPACKET
		
		upStreamOctets = Long.parseLong(items[10]);
		upStreamPacket = Long.parseLong(items[11]);
		
		if(!ports.containsKey(destport))
		{
			return ret;
		}
		
		BasicTuple tuple = new BasicTuple();
		tuple.setDestIP(destip);
		tuple.setSourceIP(sourceip);
		tuple.setDestPort(String.valueOf(destport));
		tuple.setGIS(gisstr);
		tuple.setUpStreamOctets(upStreamOctets);
		tuple.setUpStreamPacket(upStreamPacket);
		tuple.setIDC(idc);
		tuple.setSessionNum(1L);
		tuple.setFlowDirection(flowdirection);
		
		String key = String.format("%s_%s_%s", destip,destport,sourceip);
		
		ret.setFields(key, tuple);
		return ret;
		    
	}

}

    